package bold.module.kafka;

import lombok.extern.log4j.Log4j2;
import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.util.Arrays;
import java.util.Properties;

@Log4j2
public class Consumer extends Thread {

    private KafkaConsumer kafkaConsumer;

    public Consumer() {
        Properties props = new Properties();
        props.put("bootstrap.servers", Config.brokerList);
        props.put("group.id", Config.groupId);
        //自动提交位移
        props.put("enable.auto.commit", "true");
        props.put("auto.commit.interval.ms", "1000");
        props.put("session.timeout.ms", "30000");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

        //需要鉴权
        if (false) {
            props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_PLAINTEXT");
            props.put("sasl.kerberos.service.name", "ocdp");
            props.put("sasl.mechanism", "GSSAPI");
        }
        this.kafkaConsumer = new KafkaConsumer<String, String>(props);
        kafkaConsumer.subscribe(Arrays.asList(Config.topic));
    }

    @Override
    public void run() {
        while (true) {
            // 使用200ms作为获取超时时间
            ConsumerRecords<String, String> records = kafkaConsumer.poll(200);

            for (ConsumerRecord<String, String> record : records) {
                // 简单地打印消息
                /* System.out.println("===" + record.value() + " consumed " + record.partition() + " message with offset: " + record.offset());
                 */
                log.debug("获取：" + record.value());
            }
        }
    }

    public static void main(String[] args) {
        new Consumer().start();
    }
}
